package com.topsun.framework.common.kafka.consumer;

import com.alibaba.fastjson2.JSONObject;
import com.topsun.framework.common.kafka.pojo.Message;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaHandler;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import sun.reflect.generics.reflectiveObjects.ParameterizedTypeImpl;

import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.Iterator;
import java.util.Objects;

/**
 * @Package cloud.chenglang.data.kafka
 * @author: 皮皮
 * @date: 2021/7/7 5:34 PM
 */
@Slf4j
@SuppressWarnings("all")
public abstract class DefaultConsumer<K, V> {

    @KafkaHandler
    public final void message(String message, @Header(KafkaHeaders.RAW_DATA) ConsumerRecord<K, String> consumerRecord, @Header(KafkaHeaders.GROUP_ID) String groupId, Acknowledgment ack) {
        // 判断消息是否丢弃
        if (filterMessage(consumerRecord)) {
            ack.acknowledge();
            return;
        }
        log.debug("topic = {}, groupId = {}, offset = {}, message = {}", consumerRecord.topic(), groupId, consumerRecord.offset(), consumerRecord.value());
        try {
            if (message(consumerRecord)) {
                ack.acknowledge();
            }
        } catch (Exception e) {
            log.error("topic = {}, groupId = {}, offset = {}, message = {}", consumerRecord.topic(), groupId, consumerRecord.offset(), consumerRecord.value(), e);
            throw e;
        }
    }

    /**
     * 过滤消息
     * @param consumerRecord
     * @return true:消息丢弃，false:消息处理
     */
    protected boolean filterMessage(ConsumerRecord<K, String> consumerRecord) {
        return false;
    }

    /**
     * 处理消息
     * @param consumerRecord
     * @return
     */
    public boolean message(ConsumerRecord<K, String> consumerRecord) {
        Message<K, V> message = parseMessage(consumerRecord.value());
        if (Objects.isNull(message)) {
            return true;
        }
        message.setKey(consumerRecord.key());
        Iterator<org.apache.kafka.common.header.Header> iterator = consumerRecord.headers().iterator();
        for (;iterator.hasNext();) {
            org.apache.kafka.common.header.Header header = iterator.next();
            message.getHeaders().put(header.key(), new String(header.value()));
        }
        return processMessage(consumerRecord.topic(), message);
    }

    /**
     * 解析消息
     * @param message
     * @return
     */
    private Message<K, V> parseMessage(String message) {
        if (Objects.isNull(message)) {
            return null;
        }
        Type[] types = getMessageGenericType();
        try {
            Object data = JSONObject.parseObject(message, ParameterizedTypeImpl.make(Message.class, types, Message.class));
            return (Message<K, V>) data;
        } catch (Exception e) {
            log.warn("message parse error, messageType = {}, message = {}", types, message);
        }
        return null;
    }

    /**
     * 获取泛型数据类型
     * @return
     */
    private Type[] getMessageGenericType() {
        Type superType = this.getClass().getGenericSuperclass();
        return (superType instanceof ParameterizedType ? ((ParameterizedType)superType).getActualTypeArguments() : new Type[]{});
    }

    /**
     * 处理消息
     * @param topic
     * @param data
     * @return
     */
    protected boolean processMessage(String topic, Message<K, V> data) {
        return processMessage(data);
    }

    /**
     * 处理消息
     * @param data
     * @return
     */
    protected abstract boolean processMessage(Message<K, V> data);
}
